package com.kevin.utilits.zookeeper;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang.math.RandomUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.test.TestingServer;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;

import com.google.common.base.Preconditions;

/**
 *********************************************** 
 * Copyright (c) 2016 kevin.COM All Rights Reserved. 
 * FileName：com.kevin..utilits.zookeeper.CuratorUtils.java 
 * Created On: 2016年7月7日 下午5:56:45 
 * Description:
 * @author kevin02.liu
 * @version 1.0
 *********************************************** 
 */
public class CuratorUtilsTest {
	private static int availableProcessors = Runtime.getRuntime().availableProcessors();
	private static Logger logger = Logger.getLogger(CuratorUtilsTest.class);
	private static CuratorFramework client;

	public void persistEphemeralSequential(final String key) {
		try {
			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(key);
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
			RegExceptionHandler.handleException(ex);
		}
	}

	public void remove(final String key) {
		try {
			client.delete().deletingChildrenIfNeeded().forPath(key);
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
			RegExceptionHandler.handleException(ex);
		}
	}

	public long getRegistryCenterTime(final String key) {
		long result = 0L;
		try {
			String path = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(key);
			result = client.checkExists().forPath(path).getCtime();
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
			RegExceptionHandler.handleException(ex);
		}
		Preconditions.checkState(0L != result, "Cannot get registry center time.");
		return result;
	}

	public static CuratorFramework getClient() {
		return client;
	}

	public static void setClient(CuratorFramework client) {
		CuratorUtilsTest.client = client;
	}

	public static void NodeCache(String path){
		try{
			final NodeCache nodeCache = new NodeCache(client, path, false);
			nodeCache.start();
			nodeCache.getListenable().addListener(new NodeCacheListener() {
				
				public void nodeChanged() throws Exception {
					logger.info(nodeCache.getCurrentData());
				}
			});
			nodeCache.close();
		}catch(Exception e){
			RegExceptionHandler.handleException(e);
		}
	}
	
	public static void PathChildrenCache(String path){
		try{
			PathChildrenCache cache = new PathChildrenCache(client, path, true);
			cache.getListenable().addListener(new PathChildrenCacheListener() {
				public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
					switch(event.getType()){
						case CHILD_ADDED: logger.info(event.getData().getPath());break;
						case CHILD_UPDATED: logger.info(event.getData().getPath());break;
						case CHILD_REMOVED: logger.info(event.getData().getPath()); break;
						default:break;
					}
				}
			});
			cache.start(StartMode.POST_INITIALIZED_EVENT);
			cache.close();
		}catch(Exception e){
			logger.error(e.getMessage(), e);
		} 
	}
	public static void testLeaderSelector(){
		TestingServer server = null;
		try {
			server = new TestingServer(2181, true);
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
		CuratorUtils.init("127.0.0.1:2181", 10, 3, 10, "root");
		ExecutorService fixedThreadPool = Executors.newFixedThreadPool(availableProcessors);  
		for(int i=0; i< availableProcessors; i++){
			fixedThreadPool.submit(new Runnable(){
				public void run(){
					try{
						final LeaderSelector leaderSelector = new LeaderSelector(client, "/testLeaderSelector", new LeaderSelectorListener() {
							public void stateChanged(CuratorFramework arg0, ConnectionState arg1) {
								logger.info("stateChanged name=" + Thread.currentThread().getName());
							}
							public void takeLeadership(CuratorFramework client) throws Exception {
								logger.info("takeLeadership name=" + Thread.currentThread().getName());
							}
						}); 
						leaderSelector.autoRequeue();
						leaderSelector.start();
						leaderSelector.close();
				     }catch(Exception e){  
				    	 logger.error(e.getMessage(), e);
				     }  
				}
			});
		}
		
		CuratorUtils.close();
		fixedThreadPool.shutdown();
		if(server != null){
			try {
				server.stop();
			} catch (IOException e) {
				logger.error(e.getMessage(), e);
			}
		}
	}
	static int l = 0;
	public static void testDistributedLock(){
		CuratorUtils.init("10.199.195.179:2181", 10, 3, 10, "root");
		ExecutorService fixedThreadPool = Executors.newFixedThreadPool(availableProcessors);  
		for(int i=0; i< availableProcessors*10; i++){
			fixedThreadPool.submit(new Runnable(){
				public void run(){
					try{
						InterProcessMutex lock = new InterProcessMutex(client, "/testDistributedLock");
						lock.acquire();
						logger.info("acquire");
						l += 1000;
						logger.info("l=" + l);
						lock.release();
						logger.info("release done");
				     }catch(Exception e){  
				    	 logger.error(e.getMessage(), e);
				     }  
				}
			});
		}
		
		CuratorUtils.close();
		fixedThreadPool.shutdown();
	}
	
	public static void testDistributedBarrier(){
		CuratorUtils.init("10.199.195.179:2181", 10, 3, 10, "root");
		final DistributedBarrier barrier = new DistributedBarrier(client, "/testDistributedBarrier");
		ExecutorService fixedThreadPool = Executors.newFixedThreadPool(availableProcessors);  
		for(int i=0; i< availableProcessors; i++){
			fixedThreadPool.submit(new Runnable(){
				public void run(){
					try{
						barrier.setBarrier();
						logger.info("waitOnBarrier");
						barrier.waitOnBarrier();//等待barrier为0
						logger.info("waitOnBarrier done");
				     }catch(Exception e){  
				    	 logger.error(e.getMessage(), e);
				     }  
				}
			});
		}
		try {
			barrier.removeBarrier();//清空计数器
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
		
		CuratorUtils.close();
		fixedThreadPool.shutdown();
	}
	public static void testDistributedDoubleBarrier(){
		CuratorUtils.init("10.199.195.179:2181", 10, 3, 10, "root");
		ExecutorService fixedThreadPool = Executors.newFixedThreadPool(availableProcessors);  
		for(int i=0; i< availableProcessors; i++){
			fixedThreadPool.submit(new Runnable(){
				public void run() {
					try {
						DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(client, "/distributedDoubleBarrier", 5);
						barrier.enter();
						logger.info(Thread.currentThread().getName() + " enter ");
						barrier.leave();//清空计数器
						logger.info(Thread.currentThread().getName() + " leave ");
						Thread.sleep(RandomUtils.nextInt(100));  
				     } catch (Exception e) {  
				    	logger.error(e.getMessage(), e);
				     }  
				}
			});
		}
		CuratorUtils.close();
		fixedThreadPool.shutdown();
	}
	
	public static void main(String[] args){
		testLeaderSelector();
	}
	public static void testDistributedInteger(){
		CuratorUtils.init("10.199.195.179:2181", 10, 3, 10, "root");
		ExecutorService fixedThreadPool = Executors.newFixedThreadPool(availableProcessors);  
		for(int i=0; i< availableProcessors; i++){
			fixedThreadPool.submit(new Runnable(){
				public void run() {
					try{
						CuratorUtils.distributedInteger("/distributedInt", 1000);
						Thread.sleep(RandomUtils.nextInt(100));
				     }catch (InterruptedException e) {
				    	 logger.error(e.getMessage(), e);
				     }
				}
			});
		}
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			logger.error(e.getMessage(), e);
		}
		CuratorUtils.close();
		fixedThreadPool.shutdown();
	}
}
